package com.bbxdemo.kafka.produce;

import kafka.cluster.Partition;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Properties;

/**
 * 生产优化，可优化内容如下,
 * 属性       默认值
 * batch.size:16384
 * linger.ms:0   实时发送消息
 * buffer.memory:32 * 1024 * 1024
 * compression.type:none
 */
@Slf4j
@RestController
@RequestMapping("/produce/set")
public class SetProduceDemo {


    /**
     * 提高生产吞吐
     *
     * @param msg
     */
    @RequestMapping("/config")
    public void config(String msg) throws InterruptedException {
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //优化product 参数，
        //缓冲区大小
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32 * 1024 * 1024 * 1024L);
        //批大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 2638400);
        //
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10000);

        //压缩类型
        // <code>none</code>,
        // <code>gzip</code>,
        // <code>snappy</code>,
        // <code>lz4</code>, or
        // <code>zstd</code>. "
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        log.info("start to  seng Msg [{}]", System.currentTimeMillis());
        for (int i = 0; i < 10; i++) {

            //创建发送的内容
            ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic", 0, "",
                    msg + "-" + i);
            //发送数据
            producer.send(producerRecord);
        }
        log.info("end  seng Msg [{}]", System.currentTimeMillis());

        //此处通过让主线程 sleep 来观察修改参数后的实际效果
        //修改前能够实时收到消息，修改后，只有满足 batch.size  或者linger.ms(默认为0，因此会立即发送) 才发送
        Thread.sleep(60000);
        log.info("end   sleep[{}]", System.currentTimeMillis());
        //关闭资源
        producer.close();
    }


    /**
     * 设置应答，应答策略如下
     * ack 可选参数 ,默认值为all
     * "all", "-1", "0", "1"
     * -1：不需要应答,
     * 0：leader收到落后应答返回
     * all、1：leader 和isr 队列中所有节点全部收到后应答返回，
     * leader 维护了一个动态的 in-sync replica  set (isr)---与leader 进行定时通讯的 follower ，
     * 例如（leader：0，isr：0、1、2）表示 leader 节点是0， 活跃的follower 节点是   0，1，2，
     * 定时通讯的时间阈值 由 replica.lag.time.max.ms设定 ，默认值为 30
     *
     * 数据完全可靠的条件= ack为1 + 分区副本大于等于2 + ISR 中最小副本数大于等于2
     * min.insync.replicas: ISR 最小副本数，默认为1
     *
     * @param msg
     */
    @RequestMapping("/setAck")
    public void setAck(String msg) throws InterruptedException {
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置应答策略
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        //设置重试次数，默认值为int 最大值
        properties.put(ProducerConfig.RETRIES_CONFIG, 2);

        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        log.info("start to  seng Msg [{}]", System.currentTimeMillis());
        for (int i = 0; i < 100000; i++) {
            //创建发送的内容
            ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic-200000", msg + "-" + i);
            //发送数据
            producer.send(producerRecord);
        }
        producer.close();
    }


    /**
     * 开启幂等
     * @param msg
     * @throws InterruptedException
     */
    @RequestMapping("/idempotence")
    public void idempotence(String msg) throws InterruptedException {
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置幂等性 ，单会话 单分区  幂等 默认为true
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        log.info("start to  seng Msg [{}]", System.currentTimeMillis());
        for (int i = 0; i < 10; i++) {
            //创建发送的内容
            ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic", msg + "-" + i);
            //发送数据
            producer.send(producerRecord);
        }
        producer.close();
    }
    /**
     * 事务消息
     *
     *
     * @param msg
     * @throws InterruptedException
     */

    @RequestMapping("/transactional")
    public void transactional(String msg) throws InterruptedException {
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //开始事务，需要设定事务ID，事务ID 保证全局唯一
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "222222222");

        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //初始化 事务
        producer.initTransactions();
        //开启事务
        producer.beginTransaction();
        log.info("start to  seng Msg [{}]", System.currentTimeMillis());
        try {
            for (int i = 0; i < 10; i++) {
                //创建发送的内容
                ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic", 0, "",
                        msg + "-" + i);
                //发送数据
                producer.send(producerRecord);
                //故意报错   测试事务，不会收到消息
                if(i==5){
                   i= i / 0;
                }
            }
            log.info("end  seng Msg [{}]", System.currentTimeMillis());

        } catch (Exception e) {
            //终止事务
            producer.abortTransaction();
        } finally {
            log.info("end   sleep[{}]", System.currentTimeMillis());
            //关闭资源
            producer.close();
        }



    }


}
